/**
 * Copyright (c) "Neo4j"
 * Neo4j Sweden AB [https://neo4j.com]
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import { Notification, of, throwError, firstValueFrom } from 'rxjs'
import { map, materialize, toArray, concatWith } from 'rxjs/operators'
import neo4j, { ResultSummary, int } from '../../src'
import RxSession from '../../src/session-rx'
import sharedNeo4j from '../internal/shared-neo4j'
import {
  newError,
  error,
  internal,
  Session,
  ConnectionProvider
} from 'neo4j-driver-core'

const { SERVICE_UNAVAILABLE, SESSION_EXPIRED } = error
const { bookmarks, logger, constants: { TELEMETRY_APIS } } = internal

describe('#integration rx-session', () => {
  let driver
  /** @type {RxSession} */
  let session
  /** @type {number} */
  let protocolVersion

  beforeEach(async () => {
    driver = neo4j.driver(
      `bolt://${sharedNeo4j.hostnameWithBoltPort}`,
      sharedNeo4j.authToken
    )
    session = driver.rxSession()

    protocolVersion = await sharedNeo4j.cleanupAndGetProtocolVersion(driver)
  })

  afterEach(async () => {
    if (session) {
      await session.close().toPromise()
    }
    await driver.close()
  })

  it('should be able to run a simple query', async () => {
    if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
      return
    }

    const result = await session
      .run('UNWIND [1,2,3,4] AS n RETURN n')
      .records()
      .pipe(
        map(r => r.get('n').toInt()),
        materialize(),
        toArray()
      )
      .toPromise()

    expect(result).toEqual([
      Notification.createNext(1),
      Notification.createNext(2),
      Notification.createNext(3),
      Notification.createNext(4),
      Notification.createComplete()
    ])
  }, 60000)

  it('should be able to reuse session after failure', async () => {
    if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
      return
    }

    const result1 = await session
      .run('INVALID STATEMENT')
      .records()
      .pipe(materialize(), toArray())
      .toPromise()
    expect(result1[0].error.message).toMatch(/Invalid input/)

    const result2 = await session
      .run('RETURN 1')
      .records()
      .pipe(
        map(r => r.get(0).toInt()),
        materialize(),
        toArray()
      )
      .toPromise()
    expect(result2).toEqual([
      Notification.createNext(1),
      Notification.createComplete()
    ])
  }, 60000)

  it('should run transactions without retries', async () => {
    if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
      return
    }

    const txcWork = new ConfigurableTransactionWork({
      query: 'CREATE (:WithoutRetry) RETURN 5'
    })

    const result = await session
      .executeWrite(txc => txcWork.work(txc))
      .pipe(materialize(), toArray())
      .toPromise()
    expect(result).toEqual([
      Notification.createNext(5),
      Notification.createComplete()
    ])

    expect(txcWork.invocations).toBe(1)
    expect(await countNodes('WithoutRetry')).toBe(1)
  }, 60000)

  it('should run transaction with retries on reactive failures', async () => {
    if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
      return
    }

    const txcWork = new ConfigurableTransactionWork({
      query: 'CREATE (:WithReactiveFailure) RETURN 7',
      reactiveFailures: [
        newError('service unavailable', SERVICE_UNAVAILABLE),
        newError('session expired', SESSION_EXPIRED),
        newError('transient error', 'Neo.TransientError.Transaction.NotStarted')
      ]
    })

    const result = await session
      .executeWrite(txc => txcWork.work(txc))
      .pipe(materialize(), toArray())
      .toPromise()
    expect(result).toEqual([
      Notification.createNext(7),
      Notification.createComplete()
    ])

    expect(txcWork.invocations).toBe(4)
    expect(await countNodes('WithReactiveFailure')).toBe(1)
  }, 60000)

  it('should run transaction with retries on synchronous failures', async () => {
    if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
      return
    }

    const txcWork = new ConfigurableTransactionWork({
      query: 'CREATE (:WithSyncFailure) RETURN 9',
      syncFailures: [
        newError('service unavailable', SERVICE_UNAVAILABLE),
        newError('session expired', SESSION_EXPIRED),
        newError('transient error', 'Neo.TransientError.Transaction.NotStarted')
      ]
    })

    const result = await session
      .executeWrite(txc => txcWork.work(txc))
      .pipe(materialize(), toArray())
      .toPromise()
    expect(result).toEqual([
      Notification.createNext(9),
      Notification.createComplete()
    ])

    expect(txcWork.invocations).toBe(4)
    expect(await countNodes('WithSyncFailure')).toBe(1)
  }, 60000)

  it('should fail on transactions that cannot be retried', async () => {
    if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
      return
    }

    const txcWork = new ConfigurableTransactionWork({
      query: 'UNWIND [10, 5, 0] AS x CREATE (:Hi) RETURN 10/x'
    })

    const result = await session
      .executeWrite(txc => txcWork.work(txc))
      .pipe(materialize(), toArray())
      .toPromise()
    expect(result[0].value).toBe(1)
    expect(result[1].value).toBe(2)
    expect(result[2].error.message).toMatch(/\/ by zero/)

    expect(txcWork.invocations).toBe(1)
    expect(await countNodes('Hi')).toBe(0)
  }, 60000)

  it('should fail even after a transient error', async () => {
    if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
      return
    }

    const txcWork = new ConfigurableTransactionWork({
      query: 'CREATE (:Person) RETURN 1',
      syncFailures: [
        newError(
          'a transient error',
          'Neo.TransientError.Transaction.NotStarted'
        )
      ],
      reactiveFailures: [
        newError('a database error', 'Neo.Database.Not.Started')
      ]
    })

    const result = await session
      .executeWrite(txc => txcWork.work(txc))
      .pipe(materialize(), toArray())
      .toPromise()
    expect(result[0].error.message).toMatch(/a database error/)

    expect(txcWork.invocations).toBe(2)
    expect(await countNodes('Person')).toBe(0)
  }, 60000)

  describe('.executeWrite()', () => {
    it('should run transactions without retries', async () => {
      if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
        return
      }

      const txcWork = new ConfigurableTransactionWork({
        query: 'CREATE (:WithoutRetry) RETURN 5'
      })

      const result = await session
        .executeWrite(txc => txcWork.work(txc))
        .pipe(materialize(), toArray())
        .toPromise()
      expect(result).toEqual([
        Notification.createNext(5),
        Notification.createComplete()
      ])

      expect(txcWork.invocations).toBe(1)
      expect(await countNodes('WithoutRetry')).toBe(1)
    }, 60000)

    it('should run transaction with retries on reactive failures', async () => {
      if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
        return
      }

      const txcWork = new ConfigurableTransactionWork({
        query: 'CREATE (:WithReactiveFailure) RETURN 7',
        reactiveFailures: [
          newError('service unavailable', SERVICE_UNAVAILABLE),
          newError('session expired', SESSION_EXPIRED),
          newError('transient error', 'Neo.TransientError.Transaction.NotStarted')
        ]
      })

      const result = await session
        .executeWrite(txc => txcWork.work(txc))
        .pipe(materialize(), toArray())
        .toPromise()
      expect(result).toEqual([
        Notification.createNext(7),
        Notification.createComplete()
      ])

      expect(txcWork.invocations).toBe(4)
      expect(await countNodes('WithReactiveFailure')).toBe(1)
    }, 60000)

    it('should run transaction with retries on synchronous failures', async () => {
      if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
        return
      }

      const txcWork = new ConfigurableTransactionWork({
        query: 'CREATE (:WithSyncFailure) RETURN 9',
        syncFailures: [
          newError('service unavailable', SERVICE_UNAVAILABLE),
          newError('session expired', SESSION_EXPIRED),
          newError('transient error', 'Neo.TransientError.Transaction.NotStarted')
        ]
      })

      const result = await session
        .executeWrite(txc => txcWork.work(txc))
        .pipe(materialize(), toArray())
        .toPromise()
      expect(result).toEqual([
        Notification.createNext(9),
        Notification.createComplete()
      ])

      expect(txcWork.invocations).toBe(4)
      expect(await countNodes('WithSyncFailure')).toBe(1)
    }, 60000)

    it('should fail on transactions that cannot be retried', async () => {
      if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
        return
      }

      const txcWork = new ConfigurableTransactionWork({
        query: 'UNWIND [10, 5, 0] AS x CREATE (:Hi) RETURN 10/x'
      })

      const result = await session
        .executeWrite(txc => txcWork.work(txc))
        .pipe(materialize(), toArray())
        .toPromise()
      expect(result[0].value).toBe(1)
      expect(result[1].value).toBe(2)
      expect(result[2].error.message).toMatch(/\/ by zero/)

      expect(txcWork.invocations).toBe(1)
      expect(await countNodes('Hi')).toBe(0)
    }, 60000)

    it('should fail even after a transient error', async () => {
      if (protocolVersion.isLessThan({ major: 4, minor: 0 })) {
        return
      }

      const txcWork = new ConfigurableTransactionWork({
        query: 'CREATE (:Person) RETURN 1',
        syncFailures: [
          newError(
            'a transient error',
            'Neo.TransientError.Transaction.NotStarted'
          )
        ],
        reactiveFailures: [
          newError('a database error', 'Neo.Database.Not.Started')
        ]
      })

      const result = await session
        .executeWrite(txc => txcWork.work(txc))
        .pipe(materialize(), toArray())
        .toPromise()
      expect(result[0].error.message).toMatch(/a database error/)

      expect(txcWork.invocations).toBe(2)
      expect(await countNodes('Person')).toBe(0)
    }, 60000)
  })

  async function countNodes (label) {
    const session = driver.rxSession()
    return await session
      .run(`MATCH (n:${label}) RETURN count(n)`)
      .records()
      .pipe(
        map(r => r.get(0).toInt()),
        concatWith(session.close())
      )
      .toPromise()
  }
  class ConfigurableTransactionWork {
    constructor ({ query, syncFailures = [], reactiveFailures = [] } = {}) {
      this._query = query
      this._syncFailures = syncFailures
      this._syncFailuresIndex = 0
      this._reactiveFailures = reactiveFailures
      this._reactiveFailuresIndex = 0
      this._invocations = 0
    }

    get invocations () {
      return this._invocations
    }

    work (txc) {
      this._invocations++

      if (this._syncFailuresIndex < this._syncFailures.length) {
        throw this._syncFailures[this._syncFailuresIndex++]
      }

      if (this._reactiveFailuresIndex < this._reactiveFailures.length) {
        const i = this._reactiveFailuresIndex++
        return throwError(() => this._reactiveFailures[i])
      }

      return txc
        .run(this._query)
        .records()
        .pipe(map(r => r.get(0).toInt()))
    }
  }
})

describe('#unit rx-session', () => {
  describe('lastBookmarks', () => {
    ;[
      bookmarks.Bookmarks.empty(),
      new bookmarks.Bookmarks('bookmark1'),
      new bookmarks.Bookmarks(['bookmark1', 'bookmark2'])
    ].forEach(bookmarks => {
      it(`should return ${bookmarks}`, () => {
        const session = newSession(bookmarks)
        expect(session.lastBookmarks()).toBe(bookmarks.values())
      })
    })
  })

  describe('.run()', () => {
    it('should redirect run to the underlying session', () => {
      const capture = []
      const _session = {
        run: (...args) => {
          capture.push(args)
          const summary = new ResultSummary(args[0], args[1], {}, 5.3)
          return {
            async * [Symbol.asyncIterator] () {
              return summary
            },
            async summary () {
              return summary
            }
          }
        }
      }
      const session = new RxSession({ session: _session })
      const query = 'the query'
      const params = {}
      const txConfig = { timeout: 1000.120 }

      session.run(query, params, txConfig).records().subscribe()

      expect(capture).toEqual([[query, params, txConfig]])
    })
  })

  describe('._beginTransaction()', () => {
    it('should round up sub milliseconds transaction timeouts', () => {
      const capture = []
      const _session = {
        _beginTransaction: async (...args) => {
          capture.push(args)
          return {}
        }
      }

      const session = new RxSession({
        session: _session
      })

      const accessMode = 'READ'
      const timeout = 0.2

      session._beginTransaction(accessMode, { timeout })
        .subscribe()

      expect(capture[0][1].timeout).toEqual(int(1))
    })

    it('should log a warning for timeout configurations with sub milliseconds', () => {
      const capture = []
      const loggerFunction = (...args) => capture.push(args)
      const log = new logger.Logger('debug', loggerFunction)

      const _session = {
        _beginTransaction: async (...args) => {
          return {}
        }
      }

      const session = new RxSession({
        session: _session,
        log
      })

      const accessMode = 'READ'
      const timeout = 0.2

      session._beginTransaction(accessMode, { timeout })
        .subscribe()

      expect(capture[0]).toEqual([
        'info',
        `Transaction timeout expected to be an integer, got: ${timeout}. The value will be rounded up.`
      ])
    })

    it('should not log a warning for timeout configurations without sub milliseconds', () => {
      const capture = []
      const loggerFunction = (...args) => capture.push(args)
      const log = new logger.Logger('debug', loggerFunction)

      const _session = {
        _beginTransaction: async (...args) => {
          return {}
        }
      }

      const session = new RxSession({
        session: _session,
        log
      })

      const accessMode = 'READ'
      const timeout = 1

      session._beginTransaction(accessMode, { timeout })
        .subscribe()

      expect(capture.length).toEqual(0)
    })
  })

  describe('.beginTransaction()', () => {
    it(`should send telemetry configuration with API equals to ${TELEMETRY_APIS.UNMANAGED_TRANSACTION}`, () => {
      const capture = []
      const _session = {
        _beginTransaction: async (...args) => {
          capture.push(args)
          return {}
        }
      }

      const session = new RxSession({
        session: _session
      })

      const timeout = 0.2

      session.beginTransaction({ timeout })
        .subscribe()

      expect(capture[0][2]).toEqual({
        api: TELEMETRY_APIS.UNMANAGED_TRANSACTION
      })
    })
  })

  ;[
    'executeRead',
    'executeWrite'
  ].forEach(txFun => {
    describe(`.${txFun}()`, () => {
      it(`should send telemetry configuration with API equals to ${TELEMETRY_APIS.MANAGED_TRANSACTION}`, async () => {
        const capture = []
        const _session = {
          _beginTransaction: async (...args) => {
            capture.push(args)
            return {
              commit: () => Promise.resolve(),
              rollback: () => Promise.resolve(),
              run: () => Promise.resolve(),
              isOpen: () => true
            }
          }
        }

        const session = new RxSession({
          session: _session
        })

        const fun = session[txFun].bind(session)
        await firstValueFrom(fun(() => { return of(0) }))

        expect(capture.length).toEqual(1)
        expect(capture[0][2].api).toEqual(TELEMETRY_APIS.MANAGED_TRANSACTION)
      })

      it('should send telemetry on retry original when telemetry doesn\'t succeeded', async () => {
        const capture = []
        const errors = [newError('message', SERVICE_UNAVAILABLE)]
        const _session = {
          _beginTransaction: async (...args) => {
            capture.push(args)
            const error = errors.pop()
            if (error) {
              throw error
            }
            return {
              commit: () => Promise.resolve(),
              rollback: () => Promise.resolve(),
              run: () => Promise.resolve(),
              isOpen: () => true
            }
          }
        }

        const session = new RxSession({
          session: _session
        })

        const fun = session[txFun].bind(session)
        await firstValueFrom(fun(() => { return of(0) }))

        expect(capture.length).toEqual(2)
        expect(capture[0][2].api).toEqual(TELEMETRY_APIS.MANAGED_TRANSACTION)
        expect(capture[1][2].api).toEqual(TELEMETRY_APIS.MANAGED_TRANSACTION)
      })

      it('should not send telemetry on retry original when telemetry succeeded', async () => {
        const capture = []
        const errors = [newError('message', SERVICE_UNAVAILABLE)]
        const _session = {
          _beginTransaction: async (...args) => {
            capture.push(args)
            if (args[2] && args[2].onTelemetrySuccess) {
              args[2].onTelemetrySuccess()
            }
            const error = errors.pop()
            if (error) {
              throw error
            }
            return {
              commit: () => Promise.resolve(),
              rollback: () => Promise.resolve(),
              run: () => Promise.resolve(),
              isOpen: () => true
            }
          }
        }

        const session = new RxSession({
          session: _session
        })

        const fun = session[txFun].bind(session)
        await firstValueFrom(fun(() => { return of(0) }))

        expect(capture.length).toEqual(2)
        expect(capture[0][2].api).toEqual(TELEMETRY_APIS.MANAGED_TRANSACTION)
        expect(capture[1][2]).toEqual(undefined)
      })
    })
  })

  function newSession (lastBookmarks = bookmarks.Bookmarks.empty()) {
    const connectionProvider = new ConnectionProvider()
    connectionProvider.acquireConnection = () => Promise.resolve(null)
    connectionProvider.close = () => Promise.resolve()

    const session = new Session({
      mode: 'READ',
      connectionProvider,
      database: '',
      config: {},
      reactive: true,
      bookmarks: lastBookmarks
    })

    return new RxSession({ session })
  }
})
